{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Python 容器使用的 5 个技巧和 2 个误区\n",
    "\n",
    "https://yq.aliyun.com/articles/719141?spm=a2c4e.11155472.0.0.5c217f8fqXNF7H\n",
    "\n",
    "## 避免频繁扩充列表/创建新列表\n",
    "\n",
    "- 更多的使用 yield 关键字，返回生成器对象\n",
    "\n",
    "- 尽量使用生成器表达式替代列表推导表达式\n",
    "\n",
    "    生成器表达式： (i for in range(100))\n",
    "    \n",
    "    列表推导表达式： [i for in range(100)]\n",
    "    \n",
    "- 尽量使用模块提供的懒惰对象：\n",
    "\n",
    "    使用 re.finditer 替代 re.findall\n",
    "    \n",
    "    直接使用可迭代的文件对象： for line in fp，而不是 for line in fp.readlines()\n",
    "    \n",
    "## 在列表头部操作多的场景使用 deque 模块\n",
    "\n",
    "## 使用集合/字典来判断成员是否存在"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 面向容器接口编程"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def add_ellipsis_gen(comments, max_length=12):\n",
    "    \"\"\"如果可迭代评论里的内容超过 max_length，剩下的字符用省略号代替\n",
    "    \"\"\"\n",
    "    for comment in comments:\n",
    "        comment = comment.strip()\n",
    "        if len(comment) > max_length:\n",
    "            yield comment[:max_length] + '...'\n",
    "        else:\n",
    "            yield comment\n",
    "\n",
    "comments = [\n",
    "    \"Implementation note\",\n",
    "    \"Changed\",\n",
    "    \"ABC for generator\",\n",
    "]\n",
    "print(\"\\n\".join(add_ellipsis_gen(comments)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 使用元组改善分支代码"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 低级写法\n",
    "import time\n",
    " \n",
    "\n",
    "def from_now(ts):\n",
    "    \"\"\"接收一个过去的时间戳，返回距离当前时间的相对时间文字描述\n",
    "    \"\"\"\n",
    "    now = time.time()\n",
    "    seconds_delta = int(now - ts)\n",
    "    if seconds_delta < 1:\n",
    "        return \"less than 1 second ago\"\n",
    "    elif seconds_delta < 60:\n",
    "        return \"{} seconds ago\".format(seconds_delta)\n",
    "    elif seconds_delta < 3600:\n",
    "        return \"{} minutes ago\".format(seconds_delta // 60)\n",
    "    elif seconds_delta < 3600 * 24:\n",
    "        return \"{} hours ago\".format(seconds_delta // 3600)\n",
    "    else:\n",
    "        return \"{} days ago\".format(seconds_delta // (3600 * 24))\n",
    "\n",
    "\n",
    "now = time.time()\n",
    "print(from_now(now))\n",
    "print(from_now(now - 24))\n",
    "print(from_now(now - 600))\n",
    "print(from_now(now - 7500))\n",
    "print(from_now(now - 87500))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 高级写法\n",
    "\n",
    "import bisect\n",
    "\n",
    "\n",
    "# BREAKPOINTS 必须是已经排好序的，不然无法进行二分查找\n",
    "BREAKPOINTS = (1, 60, 3600, 3600 * 24)\n",
    "TMPLS = (\n",
    "    # unit, template\n",
    "    (1, \"less than 1 second ago\"),\n",
    "    (1, \"{units} seconds ago\"),\n",
    "    (60, \"{units} minutes ago\"),\n",
    "    (3600, \"{units} hours ago\"),\n",
    "    (3600 * 24, \"{units} days ago\"),\n",
    ")\n",
    "\n",
    "def from_now(ts):\n",
    "    \"\"\"接收一个过去的时间戳，返回距离当前时间的相对时间文字描述\n",
    "    \"\"\"\n",
    "    seconds_delta = int(time.time() - ts)\n",
    "    unit, tmpl = TMPLS[bisect.bisect(BREAKPOINTS, seconds_delta)]\n",
    "    return tmpl.format(units=seconds_delta // unit)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 在更多地方使用动态解包\n",
    "\n",
    " Python 3.5 以后的版本，你可以直接用 ** 运算符来快速完成字典的合并操作"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "user = {**{\"name\": \"piglei\"}, **{\"movies\": [\"Fight Club\"]}}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##  最好不用“获取许可”，也无需“要求原谅”\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# AF: Ask for Forgiveness\n",
    "# 要做就做，如果抛出异常了，再处理异常\n",
    "def counter_af(l):\n",
    "    result = {}\n",
    "    for key in l:\n",
    "        try:\n",
    "            result[key] += 1\n",
    "        except KeyError:\n",
    "            result[key] = 1\n",
    "    return result\n",
    "\n",
    "\n",
    "# AP: Ask for Permission\n",
    "# 做之前，先问问能不能做，可以做再做\n",
    "def counter_ap(l):\n",
    "    result = {}\n",
    "    for key in l:\n",
    "        if key in result:\n",
    "            result[key] += 1\n",
    "        else:\n",
    "            result[key] = 1\n",
    "    return result"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "整个 Python 社区对第一种 Ask for Forgiveness 的异常捕获式编程风格有着明显的偏爱。这其中有很多原因，首先，在 Python 中抛出异常是一个很轻量的操作。其次，第一种做法在性能上也要优于第二种，因为它不用在每次循环的时候都做一次额外的成员检查。\n",
    "\n",
    "如果你想统计次数的话，直接用 collections.defaultdict 就可以了："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from collections import defaultdict\n",
    " \n",
    "\n",
    "def counter_by_collections(l):\n",
    "    result = defaultdict(int)\n",
    "    for key in l:\n",
    "        result[key] += 1\n",
    "    return result"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "这样的代码既不用“获取许可”，也无需“请求原谅”。整个代码的控制流变得更清晰自然了。所以，如果可能的话，请尽量想办法省略掉那些非核心的异常捕获逻辑。一些小提示：\n",
    "\n",
    "- 操作字典成员时：使用 collections.defaultdict 类型\n",
    "\n",
    " 或者使用 dict[key]=dict.setdefault(key,0)+1 内建函数\n",
    "\n",
    "- 如果移除字典成员，不关心是否存在：\n",
    "\n",
    "- 调用 pop 函数时设置默认值，比如 dict.pop(key,None)\n",
    "- 在字典获取成员时指定默认值： dict.get(key, default_value)\n",
    "- 对列表进行不存在的切片访问不会抛出 IndexError 异常： \"foo\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 使用 next() 函数,查找第一个符合的元素\n",
    "\n",
    "next() 是一个非常实用的内建函数，它接收一个迭代器作为参数，然后返回该迭代器的下一个元素。使用它配合生成器表达式，可以高效的实现“从列表中查找第一个满足条件的成员”之类的需求。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "numbers = [3, 7, 8, 2, 21]\n",
    "# 获取并 **立即返回** 列表里的第一个偶数\n",
    "print(next(i for i in numbers if i % 2 == 0))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 使用有序字典来去重\n",
    "\n",
    "字典和集合的结构特点保证了它们的成员不会重复，所以它们经常被用来去重。但是，使用它们俩去重后的结果会**丢失原有列表的顺序**。这是由底层数据结构“哈希表（Hash Table）”的特点决定的。\n",
    "\n",
    "如果**既需要去重又必须保留顺序**怎么办？我们可以使用 `collections.OrderedDict` 模块\n",
    "\n",
    "Hint: 在 Python 3.6 中，默认的字典类型修改了实现方式，已经变成有序的了。并且在 Python 3.7 中，该功能已经从 语言的实现细节 变成了为 可依赖的正式语言特性。\n",
    "\n",
    "但是我觉得让整个 Python 社区习惯这一点还需要一些时间，毕竟目前“字典是无序的”还是被印在无数本 Python 书上。所以，我仍然建议在一切需要有序字典的地方使用 OrderedDict。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 常见误区\n",
    "\n",
    "## 当心那些已经枯竭的迭代器\n",
    "\n",
    "我们提到了使用“懒惰”生成器的种种好处。但是，所有事物都有它的两面性。生成器的最大的缺点之一就是：它会枯竭。当你完整遍历过它们后，之后的重复遍历就不能拿到任何新内容了。\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "numbers = [1, 2, 3]\n",
    "numbers = (i * 2 for i in numbers)\n",
    "\n",
    "# 第一次循环会输出 2, 4, 6\n",
    "for number in numbers:\n",
    "    print(number)\n",
    "\n",
    "# 这次循环什么都不会输出，因为迭代器已经枯竭了\n",
    "for number in numbers:\n",
    "    print(number)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "而且不光是生成器表达式，Python 3 里的 map、filter 内建函数也都有一样的特点。忽视这个特点很容易导致代码中出现一些难以察觉的 Bug。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 别在循环体内修改被迭代对象"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def remove_even(numbers):\n",
    "    \"\"\"\n",
    "    去掉列表里所有的偶数\n",
    "    \"\"\"\n",
    "    for i, number in enumerate(numbers):\n",
    "        if number % 2 == 0:\n",
    "            # 有问题的代码\n",
    "            del numbers[i]\n",
    "\n",
    "\n",
    "numbers = [1, 2, 7, 4, 8, 11]\n",
    "remove_even(numbers)\n",
    "print(numbers)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "注意到结果里那个多出来的 “8” 了吗？当你在遍历一个列表的同时修改它，就会出现这样的事情。因为被迭代的对象 numbers 在循环过程中被修改了。遍历的下标在不断增长，而列表本身的长度同时又在不断缩减。这样就会导致列表里的一些成员其实根本就没有被遍历到。\n",
    "\n",
    "所以对于这类操作，请使用一个新的空列表保存结果，或者利用 yield 返回一个生成器。而不是修改被迭代的列表或是字典对象本身。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 判断pyhon版本"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys\n",
    "PY2 = sys.version_info[0] == 2\n",
    "PY3 = sys.version_info[0] == 3"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 线程锁装饰器 "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def threadsafe_function(fcn):\n",
    "    \"\"\"decorator making sure that the decorated function is thread safe\"\"\"\n",
    "    lock = threading.RLock()\n",
    "\n",
    "    def new(*args, **kwargs):\n",
    "        \"\"\"Lock and call the decorated function\n",
    "\n",
    "           Unless kwargs['threadsafe'] == False\n",
    "        \"\"\"\n",
    "        threadsafe = kwargs.pop('threadsafe', True)\n",
    "        if threadsafe:\n",
    "            lock.acquire()\n",
    "        try:\n",
    "            ret = fcn(*args, **kwargs)\n",
    "        except Exception as excpt:\n",
    "            raise excpt\n",
    "        finally:\n",
    "            if threadsafe:\n",
    "                lock.release()\n",
    "        return ret\n",
    "    return new\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 异常处理装饰器\n",
    "\n",
    "使用轮子：pip install retry\n",
    "\n",
    "\n",
    "```python\n",
    "from retry import retry\n",
    "\n",
    "@retry(tries=3, delay=1, )\n",
    "def test():\n",
    "    for i in range(-2, 5):\n",
    "        print(1/i)\n",
    "test()\n",
    "```\n",
    "自定义："
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 参数检查装饰器\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import inspect\n",
    "\n",
    "\n",
    "def check(fn):\n",
    "    def wrapper(*args, **kwargs):\n",
    "        sig = inspect.signature(fn)\n",
    "        params = sig.parameters\n",
    "        values = list(params.values())\n",
    "        for i, p in enumerate(args):\n",
    "            param = values[i]\n",
    "            if param.annotation is not param.empty and not isinstance(\n",
    "                    p, param.annotation):\n",
    "                print(p, '!==', values[i].annotation)\n",
    "        for k, v in kwargs.items():\n",
    "            if params[k].annotation is not inspect._empty and not isinstance(\n",
    "                    v, params[k].annotation):\n",
    "                print(k, v, '!===', params[k].annotation)\n",
    "        return fn(*args, **kwargs)\n",
    "\n",
    "    return wrapper\n",
    "\n",
    "\n",
    "@check\n",
    "def add(x, y: int = 7) -> int:  #我们要求第二个参数必须是int类型,并且返回值类型也为int\n",
    "    return x + y\n",
    "\n",
    "\n",
    "print(add(2, 1))\n",
    "print(add(20, y=10))\n",
    "print(add(y=100, x=200))\n",
    "print(add(\"Yin\", \"zhengjie\"))  #我们在实际传参时故意不按照要求传参，发现会有相应的提示信息"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 异常捕获装饰器（亦可用于类方法）\n",
    "import functools\n",
    "import time\n",
    "import traceback\n",
    "\n",
    "\n",
    "def try_except_log(f=None, max_retries: int = 5, delay: (int, float) = 1, step: (int, float) = 0,\n",
    "                   exceptions: (BaseException, tuple, list) = BaseException, sleep=time.sleep,\n",
    "                   process=None, validate=None, callback=None, default=None):\n",
    "    \"\"\"\n",
    "    函数执行出现异常时自动重试的简单装饰器\n",
    "    :param f: function 执行的函数。\n",
    "    :param max_retries: int 最多重试次数。\n",
    "    :param delay: int/float 每次重试的延迟，单位秒。\n",
    "    :param step: int/float 每次重试后延迟递增，单位秒。\n",
    "    :param exceptions: BaseException/tuple/list 触发重试的异常类型，单个异常直接传入异常类型，多个异常以tuple或list传入。\n",
    "    :param sleep: 实现延迟的方法，默认为time.sleep。\n",
    "    在一些异步框架，如tornado中，使用time.sleep会导致阻塞，可以传入自定义的方法来实现延迟。\n",
    "    自定义方法函数签名应与time.sleep相同，接收一个参数，为延迟执行的时间。\n",
    "    :param process: 处理函数，函数签名应接收一个参数，每次出现异常时，会将异常对象传入。\n",
    "    可用于记录异常日志，中断重试等。\n",
    "    如处理函数正常执行，并返回True，则表示告知重试装饰器异常已经处理，重试装饰器终止重试，并且不会抛出任何异常。\n",
    "    如处理函数正常执行，没有返回值或返回除True以外的结果，则继续重试。\n",
    "    如处理函数抛出异常，则终止重试，并将处理函数的异常抛出。\n",
    "    :param validate: 验证函数，用于验证执行结果，并确认是否继续重试。\n",
    "    函数签名应接收一个参数，每次被装饰的函数完成且未抛出任何异常时，调用验证函数，将执行的结果传入。\n",
    "    如验证函数正常执行，且返回False，则继续重试，即使被装饰的函数完成且未抛出任何异常。\n",
    "    如验证函数正常执行，没有返回值或返回除False以外的结果，则终止重试，并将函数执行结果返回。\n",
    "    如验证函数抛出异常，且异常属于被重试装饰器捕获的类型，则继续重试。\n",
    "    如验证函数抛出异常，且异常不属于被重试装饰器捕获的类型，则将验证函数的异常抛出。\n",
    "    :param callback: 回调函数，函数应接收一个参数，到达重试次数且异常无法处理时，会将异常对象传入。\n",
    "    可用于记录异常日志，发送异常日志等。\n",
    "    :param default: 发生异常时， process参数处理完返回True，返回：默认值 or 默认值生成函数\n",
    "    :return: 被装饰函数的执行结果。\n",
    "    \"\"\"\n",
    "    \n",
    "    # 带参数的装饰器\n",
    "    def decorator(func):\n",
    "        @functools.wraps(func)\n",
    "        def wrapper(*args, **kwargs):\n",
    "            # nonlocal delay, step, max_retries\n",
    "            i = 0\n",
    "            func_exc, exc_traceback = None, None\n",
    "            while i < max_retries:\n",
    "                try:\n",
    "                    result = func(*args, **kwargs)\n",
    "                    # 验证函数返回False时，表示告知装饰器验证不通过，继续重试\n",
    "                    if callable(validate) and validate(result) is False:\n",
    "                        continue\n",
    "                    else:\n",
    "                        # 正常结束\n",
    "                        return result\n",
    "                except exceptions as ex:\n",
    "                    func_exc, exc_traceback = ex, traceback.format_exc()\n",
    "                    # 处理函数返回True时，表示告知装饰器异常已经处理，终止重试\n",
    "                    if callable(process):\n",
    "                        try:\n",
    "                            if process(ex) is True:\n",
    "                                return default() if callable(default) else default\n",
    "                        except Exception as e:\n",
    "                            func_exc, exc_traceback = e, traceback.format_exc()\n",
    "                            break\n",
    "                # finally:\n",
    "                i += 1\n",
    "                sleep(delay + step * i)\n",
    "            else:\n",
    "                # 回调函数，处理自动无法处理的异常\n",
    "                if callable(callback):\n",
    "                    callback(func_exc, exc_traceback)\n",
    "                return default() if callable(default) else default\n",
    "            pass\n",
    "\n",
    "        return wrapper\n",
    "\n",
    "    if callable(f):\n",
    "        return decorator(f)\n",
    "    return decorator\n",
    "\n",
    "\n",
    "@try_except_log(default=1)\n",
    "def p():\n",
    "    return 1/0\n",
    "\n",
    "\n",
    "p()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 使用goto语句\n",
    "\n",
    "```sh\n",
    "pip install goto-statement\n",
    "```\n",
    "\n",
    "具体的语法"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install goto-statement"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "from goto import with_goto\n",
    " \n",
    "@with_goto\n",
    "def range(start, stop):\n",
    "    i = start\n",
    "    result = []\n",
    " \n",
    "    label .begin\n",
    "    if i == stop:\n",
    "        goto .end\n",
    " \n",
    "    result.append(i)\n",
    "    i += 1\n",
    "    goto .begin\n",
    " \n",
    "    label .end\n",
    "    return result"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# enumerate 读取文件"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "如果要统计文件的行数，可以这样写：\n",
    "\n",
    "count = len(open(filepath, 'r').readlines())\n",
    "\n",
    "这种方法简单，但是可能比较慢，当文件比较大时甚至不能工作。\n",
    "\n",
    "可以利用enumerate()：\n",
    "```python\n",
    "count = 0\n",
    "for index, line in enumerate(open(filepath,'r'))： \n",
    "    count += 1\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 展开列表flat"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "a = [[[26, 25, 24], [23, 22, 21, 20, 19], [18, 17, 16]], [[15, 14, 13, 12, 11], [10, 9, 8], [7, 6, 5], [4, 3, 2, 1, 0], [[99, 98, 97], [96, 95, 94, 93], [92, 91, 90, 89, 88], [87, 86, 85, 84, 83]]], [[[82, 81, 80], [79, 78, 77, 76], [75, 74, 73], [72, 71, 70], [69, 68, 67]], [[66, 65, 64, 63], [62, 61, 60], [59, 58, 57, 56], [55, 54, 53, 52, 51]], [[50, 49, 48, 47, 46], [45, 44, 43, 42, 41], [40, 39, 38, 37, 36], [35, 34, 33, 32], [31, 30, 29, 28, 27]]]]\n",
    "\n",
    "# result = []\n",
    "# def flat_list(a: list):\n",
    "#     for i in a:\n",
    "#         if isinstance(i, list):\n",
    "#             flat_list(i)\n",
    "#         else:\n",
    "#             result.append(i)\n",
    "\n",
    "\n",
    "def flat_list(input_list :list):\n",
    "    result = []\n",
    "    for i in input_list:\n",
    "        result.extend([i] if not isinstance(i, list) else flat_list(i))\n",
    "    return result\n",
    "\n",
    "print(flat_list(a))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# bytes数据\n",
    "\n",
    "- bytes(iterable_of_ints) -> bytes\n",
    "- bytes(string, encoding[, errors]) -> bytes\n",
    "- bytes(bytes_or_buffer) -> immutable copy of bytes_or_buffer\n",
    "- bytes(int) -> bytes object of size given by the parameter initialized with null bytes\n",
    "- bytes() -> empty bytes object"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "a = bytes('中国', 'utf-8')\n",
    "b = bytes('abc', 'ascii')\n",
    "c = bytes(1)\n",
    "d = bytes()\n",
    "print(\"{}\\n{}\\n{}\\n{}\".format(a,b,c, d))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "list(b)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "b.hex()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "bytes.fromhex(b.hex())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 性能测试"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "True\n",
      "128 ns ± 1.86 ns per loop (mean ± std. dev. of 7 runs, 10000000 loops each)\n",
      "202 ns ± 4.89 ns per loop (mean ± std. dev. of 7 runs, 1000000 loops each)\n"
     ]
    }
   ],
   "source": [
    "a = '123abc我永远爱我的祖国'\n",
    "print(a.encode() == bytes(a, 'utf-8'))\n",
    "\n",
    "%timeit a.encode()\n",
    "\n",
    "%timeit bytes(a, 'utf-8')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# bytearray\n",
    "- bytearray(iterable_of_ints) -> bytearray\n",
    "- bytearray(string, encoding[, errors]) -> bytearray\n",
    "- bytearray(bytes_or_buffer) -> mutable copy of bytes_or_buffer\n",
    "- bytearray(int) -> bytes array of size given by the parameter initialized with null bytes\n",
    "- bytearray() -> empty bytes array"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "a = bytearray([1,2,3])\n",
    "b = bytearray(\"你好\", 'utf-8')\n",
    "c = bytearray(bytes('abc', 'ascii'))\n",
    "d = bytearray(1)\n",
    "e = bytearray()\n",
    "print(\"{}\\n{}\\n{}\\n{}\\n{}\".format(a,b,c, d, e))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "list(c)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "c.hex()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "bytearray.fromhex(c.hex())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# binascii"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import binascii"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "a = '123abc我永远爱我的祖国'\n",
    "b = a.encode()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%timeit b.hex()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%timeit binascii.hexlify(b)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "b_hex_str = b.hex()\n",
    "b_hex_str"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "binascii.unhexlify(b_hex_str).decode()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "binascii.hexlify(b)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "c = binascii.a2b_hex(b_hex_str)\n",
    "c"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# ascii & repr"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[\"'\", '1', '2', '3', 'a', 'b', 'c', '\\\\', 'u', '4', 'e', '0', '7', \"'\"]"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ascii('123abc万')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\"'123abc万'\""
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "repr(\"123abc万\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 版本"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys\n",
    "\n",
    "PY2 = sys.version_info[0] == 2\n",
    "PY3 = sys.version_info[0] == 3\n",
    "\n",
    "if PY2:\n",
    "    import Queue as queue\n",
    "    import SocketServer\n",
    "else:\n",
    "    import queue\n",
    "    import socketserver as SocketServer"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 计算大小"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys\n",
    "\n",
    "\n",
    "def get_size(obj, seen=None):\n",
    "    \"\"\"Recursively finds size of objects\"\"\"\n",
    "    size = sys.getsizeof(obj)\n",
    "    if seen is None:\n",
    "        seen = set()\n",
    "    obj_id = id(obj)\n",
    "    if obj_id in seen:\n",
    "        return 0\n",
    "    # Important mark as seen *before* entering recursion to gracefully handle\n",
    "    # self-referential objects\n",
    "    seen.add(obj_id)\n",
    "    if isinstance(obj, dict):\n",
    "        size += sum([get_size(v, seen) for v in obj.values()])\n",
    "        size += sum([get_size(k, seen) for k in obj.keys()])\n",
    "    elif hasattr(obj, '__dict__'):\n",
    "        size += get_size(obj.__dict__, seen)\n",
    "    elif hasattr(obj, '__iter__') and not isinstance(obj,\n",
    "                                                     (str, bytes, bytearray)):\n",
    "        size += sum([get_size(i, seen) for i in obj])\n",
    "    return size\n",
    "\n",
    "\n",
    "get_size(1), get_size(10000000000000000)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 图像二进制，bytes转换 (np.uint8)\n",
    "\n",
    "1. 从numpy.ndarray (图片)到字节流\n",
    "\n",
    "```python\n",
    "x = img.tobytes()\n",
    "```\n",
    "\n",
    "2. 从二进制文件到图片(numpy.ndarray)\n",
    "\n",
    "```python\n",
    "\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "cd C:\\Users\\13438\\Desktop\\"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import matplotlib.pyplot as plt\n",
    "import cv2\n",
    "from PIL import Image\n",
    "from io import BytesIO\n",
    "import numpy as np"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 图片 转 三维ndarray\n",
    "\n",
    "### 方法一: np.frombuffer/np.frombuffer + cv2.imdecode"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "with open('Image.jpg', 'rb') as f:\n",
    "    bin_str = f.read()\n",
    "#     nparr = np.frombuffer(bin_str, np.uint8) # 字符串\n",
    "    nparr = np.frombuffer(bin_str, np.uint8)\n",
    "    img_np = cv2.imdecode(nparr, cv2.IMREAD_COLOR) # cv2.CV_LOAD_IMAGE_COLOR  in CV < 3.1\n",
    "    \n",
    "    print(type(bin_str))\n",
    "    print(nparr.shape)\n",
    "    print(img_np.shape)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 方法二: cv2.imread"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "img_np = cv2.imread('Image.jpg')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 方法三 ：plt.imread"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "with open('Image.jpg', 'rb') as f:\n",
    "    img_data = plt.imread(BytesIO(f.read()),\"jpg\")\n",
    "    print(type(img_data))\n",
    "    print(img_data.shape)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 方法 四：PLT.Image\n",
    "\n",
    "略"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 三维ndarray图 转 一维ndarray\n",
    "\n",
    "encode 成一维，然后 tobytes()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 构造图片\n",
    "# img_data = np.linspace(0,255,100*100*3).reshape(100,100,-1).astype(np.uint8)\n",
    "\n",
    "# 上面的图片转 ndarray\n",
    "img_data = img_np\n",
    "ret, buf = cv2.imencode(\".jpg\", img_data)\n",
    "if ret:\n",
    "    print(img_data.shape, buf.shape)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 一维转 bytes\n",
    "接着上一步得到的一维数组"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 方法一：\n",
    "\n",
    "需要转化"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "code_folding": []
   },
   "outputs": [],
   "source": [
    "if ret:\n",
    "    img_bin1 = Image.fromarray(np.uint8(buf)).tobytes()\n",
    "    # 保存图片\n",
    "    with open('pic.jpg', 'wb') as f:\n",
    "        f.write(img_bin1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 方法二：\n",
    "\n",
    "直接转 bytes()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "img_bin2 = buf.tobytes()\n",
    "# 保存。。。\n",
    "with open('pic2.jpg', 'wb') as f:\n",
    "    f.write(img_bin2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 两种方法结果对比\n",
    "\n",
    "方法二省去了中间转换步骤，效率提高了200~300倍"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "img_bin1 == img_bin2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%timeit img_bin1 = Image.fromarray(np.uint8(buf)).tobytes()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%timeit a = buf.tobytes()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 三维转bytes\n",
    "\n",
    "先转一维再保存为bytes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "cd assets/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import cv2\n",
    "from PIL import Image\n",
    "from io import BytesIO"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 读一张数组到 三维数组\n",
    "img_data = cv2.imread('lena.jpg')\n",
    "# 转为一维\n",
    "ret, buf = cv2.imencode(\".jpg\", img_data)\n",
    "assert ret\n",
    "print(img_data.shape, buf.shape)\n",
    "# 展示\n",
    "img = Image.open(BytesIO(buf.tobytes()))\n",
    "img"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "img = Image.fromarray(cv2.cvtColor(img_data, cv2.COLOR_BGR2RGB))\n",
    "img"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 懒加载"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    " class LazyImport:\n",
    "    def __init__(self, module_name):\n",
    "        self.module_name = module_name\n",
    "        self.module = None\n",
    "        \n",
    "    def __getattr__(self, name):\n",
    "        if self.module is None:\n",
    "            self.module = __import__(self.module_name)\n",
    "        return getattr(self.module, name)\n",
    "    \n",
    "string = LazyImport(\"string\")\n",
    "print (string.digits)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 创建文件夹"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import errno\n",
    "\n",
    "\n",
    "def mkdir_p_ex(path):\n",
    "    try:\n",
    "        if os.path.exists(path) and os.path.isdir(path):\n",
    "            return True, None\n",
    "\n",
    "        os.makedirs(path)\n",
    "        return True, None\n",
    "    except OSError as exc:  # Python >2.5\n",
    "        if exc.errno == errno.EEXIST and os.path.isdir(path):\n",
    "            return True, None\n",
    "        else:\n",
    "            return False, str(exc)\n",
    "    except Exception as e:\n",
    "        return False, str(e)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 设置root日志等级"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import logging\n",
    "\n",
    "logging.root.setLevel(_CONFIG.get('INFO'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 通过UUID生成短随机ID\n",
    "\n",
    "uuid-hex长度为32，此算法缩短为22"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import string\n",
    "import uuid\n",
    "\n",
    "\n",
    "def base62(num, base=string.digits + string.ascii_letters, padding=1):\n",
    "    \"\"\"Convert a number to base-62 representation.\"\"\"\n",
    "    assert num >= 0\n",
    "    digits = []\n",
    "    while num:\n",
    "        num, rest = divmod(num, 62)\n",
    "        digits.append(base[rest])\n",
    "    digits.extend(base[0] for _ in range(len(digits), padding))\n",
    "    return ''.join(reversed(digits))\n",
    "\n",
    "\n",
    "print(base62(uuid.uuid4().int, padding=22))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 快速补全\\_\\_all__\n",
    "\n",
    "借鉴 `types`模块 源码最后一行"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "__all__ = [i for i in globals() if i[:1] != '_']"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 动态函数设计思想\n",
    "\n",
    "条件不同执行的操作不同时，且条件一般很少改变时，为了减少执行中的判断语句，可将该操作根据不同的条件执行的操作封装称一个方法，当条件改变时也将该函数重新赋值即可。\n",
    "\n",
    "例：\n",
    "\n",
    "```python\n",
    "\n",
    "if condition1:\n",
    "    操作1\n",
    "    操作2\n",
    "elif condition2:\n",
    "    操作3\n",
    "else condition3:\n",
    "    操作4\n",
    "```\n",
    "\n",
    "为减少判断，可以改成以下方式：\n",
    "\n",
    "```python\n",
    "def run1():\n",
    "    操作1\n",
    "    操作2\n",
    "    \n",
    "def run2():\n",
    "    操作3\n",
    "    \n",
    "def run3():\n",
    "    操作4\n",
    "\n",
    "# 初始执行函数（方法）\n",
    "default_run = run1\n",
    "\n",
    "def condition_change(cond):\n",
    "    global default_run\n",
    "    ...\n",
    "    # 条件改变时，切换相应的执行函数（方法）\n",
    "    default_run = run2\n",
    "\n",
    "while True:\n",
    "    # 不用判断，只管执行\n",
    "    default_run()\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 查看方法源码位置、行号"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "('<ipython-input-1-42ae88692c16>', 17)\n"
     ]
    }
   ],
   "source": [
    "import functools\n",
    "import inspect\n",
    "\n",
    "\n",
    "def _get_function_source(func):\n",
    "    func = inspect.unwrap(func)\n",
    "    if inspect.isfunction(func):\n",
    "        code = func.__code__\n",
    "        return (code.co_filename, code.co_firstlineno)\n",
    "    if isinstance(func, functools.partial):\n",
    "        return _get_function_source(func.func)\n",
    "    if isinstance(func, functools.partialmethod):\n",
    "        return _get_function_source(func.func)\n",
    "    return None\n",
    "\n",
    "\n",
    "def foo():\n",
    "    print('aaa')\n",
    "    \n",
    "print(_get_function_source(foo))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 运行异步或者同步"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "hello\n"
     ]
    }
   ],
   "source": [
    "from functools import partial\n",
    "import asyncio\n",
    "\n",
    "\n",
    "def iscoroutinefunction_or_partial(object):\n",
    "    if isinstance(object, partial):\n",
    "        object = object.func\n",
    "    return asyncio.iscoroutinefunction(object)\n",
    "\n",
    "\n",
    "def run_coroutine_or_function(func, *args, callback=None, **kwargs):\n",
    "    if iscoroutinefunction_or_partial(func):\n",
    "        f = asyncio.ensure_future(func(*args, **kwargs))\n",
    "        if callback is not None:\n",
    "            f.add_done_callback(callback)\n",
    "    else:\n",
    "        func(*args, **kwargs)\n",
    "\n",
    "\n",
    "async def test():\n",
    "    await asyncio.sleep(1)\n",
    "    print('test')\n",
    "\n",
    "run_coroutine_or_function(print, 'hello')\n",
    "run_coroutine_or_function(test)\n",
    "\n",
    "asyncio.get_event_loop().run_forever()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Http 状态码判断\n",
    "\n",
    "参考：`from httpx import codes`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def is_error(value: int) -> bool:\n",
    "    return 400 <= value <= 599\n",
    "\n",
    "\n",
    "def is_client_error(value: int) -> bool:\n",
    "    return 400 <= value <= 499\n",
    "\n",
    "\n",
    "def is_server_error(value: int) -> bool:\n",
    "    return 500 <= value <= 599"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 限制调用栈\n",
    "\n",
    "有时候，我们的某些函数可能要限制调用。例如函数 A 只能被函数 B、函数 C 调用，不能被其他函数调用。我们可以通过分析函数的调用栈来通过代码解决这个问题。查询调用栈，可以使用inspect模块的stack()函数。\n",
    "\n",
    "[详细介绍](https://mp.weixin.qq.com/s/fHpQ4NNSKkhHrQ15JUieLg)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "开始跳舞\n",
      "允许执行后续代码\n",
      "既然是跳舞，肯定要跳起来\n"
     ]
    }
   ],
   "source": [
    "import inspect\n",
    "\n",
    "\n",
    "def call_stack_check(valid_function_list=None):\n",
    "    def decorate(func):\n",
    "        def wrap(*args, **kwargs):\n",
    "            if valid_function_list:\n",
    "                stack = inspect.stack()\n",
    "                upper_function_name = stack[1].function\n",
    "                if upper_function_name not in valid_function_list:\n",
    "                    raise Exception('非法调用！')\n",
    "                else:\n",
    "                    print('允许执行后续代码')\n",
    "            result = func(*args, **kwargs)\n",
    "            return result\n",
    "        return wrap\n",
    "    return decorate\n",
    "\n",
    "\n",
    "@call_stack_check(['dance'])\n",
    "def jump():\n",
    "    print('既然是跳舞，肯定要跳起来')\n",
    "\n",
    "\n",
    "def dance():\n",
    "    print('开始跳舞')\n",
    "    jump()\n",
    "\n",
    "def sing():\n",
    "    print('开始跳舞')\n",
    "    jump()\n",
    "\n",
    "dance()  # 允许\n",
    "# sing()  # 非法"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 打印错误栈"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<module>\n",
      "main\n",
      "test2\n"
     ]
    }
   ],
   "source": [
    "import traceback\n",
    "\n",
    "\n",
    "def test2():\n",
    "    raise TypeError('abc')\n",
    "\n",
    "\n",
    "def main():\n",
    "    test2()\n",
    "\n",
    "\n",
    "try:\n",
    "    main()\n",
    "except Exception as e:\n",
    "    for frame, lineno in traceback.walk_tb(e.__traceback__):\n",
    "        print(frame.f_code.co_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 警告"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import warnings\n",
    "\n",
    "warnings.warn(\"xxx is deprecated since Python 3.8, and scheduled for removal in Python 3.10.\",\n",
    "              DeprecationWarning, stacklevel=2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 获取文件夹大小"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import re\n",
    "\n",
    "# 工具方法\n",
    "def bytes2human(n, format=\"%(value).1f%(symbol)s\"):\n",
    "    \"\"\"字节转为 M、G等单位\n",
    "    \"\"\"\n",
    "    symbols = ('B', 'K', 'M', 'G', 'T', 'P', 'E', 'Z', 'Y')\n",
    "    prefix = {}\n",
    "    for i, s in enumerate(symbols[1:]):\n",
    "        prefix[s] = 1 << (i + 1) * 10\n",
    "    for symbol in reversed(symbols[1:]):\n",
    "        if n >= prefix[symbol]:\n",
    "            value = float(n) / prefix[symbol]\n",
    "            return format % locals()\n",
    "    return format % dict(symbol=symbols[0], value=n)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## linux下通过 du -sh 命令"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_doc_usage_size_by_shell(doc_path):\n",
    "    \"\"\"\n",
    "    使用 du -sh 命令，以避免占用大小与真实大小不同\n",
    "    输入：待检测文件绝对路径\n",
    "    \"\"\"\n",
    "    response = os.popen(f'du -sh {doc_path}')\n",
    "    str_size = response.read().split()[0]\n",
    "    f_size = float(re.findall(r'[.\\d]+', str_size)[0])\n",
    "    return bytes2human(f_size)\n",
    "\n",
    "\n",
    "# 显示得更详细\n",
    "def get_doc_usage_size_by_shell2(doc_path):\n",
    "    response = os.popen(f'du -ks {doc_path}')\n",
    "    str_size = response.read().split()[0]\n",
    "    f_size = float(re.findall(r'[.\\d]+', str_size)[0])\n",
    "    return bytes2human(f_size * 1024)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 遍历所有文件累加文件大小"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_dir_real_size(dir_path):\n",
    "    size = 0\n",
    "    for root, dirs, files in os.walk(dir_path):\n",
    "        size += sum([os.path.getsize(os.path.join(root, file))\n",
    "                     for file in files])\n",
    "    return bytes2human(size)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.5"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": true
  },
  "varInspector": {
   "cols": {
    "lenName": 16,
    "lenType": 16,
    "lenVar": 40
   },
   "kernels_config": {
    "python": {
     "delete_cmd_postfix": "",
     "delete_cmd_prefix": "del ",
     "library": "var_list.py",
     "varRefreshCmd": "print(var_dic_list())"
    },
    "r": {
     "delete_cmd_postfix": ") ",
     "delete_cmd_prefix": "rm(",
     "library": "var_list.r",
     "varRefreshCmd": "cat(var_dic_list()) "
    }
   },
   "types_to_exclude": [
    "module",
    "function",
    "builtin_function_or_method",
    "instance",
    "_Feature"
   ],
   "window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
